package com.itheima.behavior.task;

import com.itheima.behavior.config.KafkaConfig;
import com.itheima.behavior.repository.BookPopularityRepository;
import com.itheima.common.redis.CacheService;
import com.itheima.model.behavior.pojos.BookPopularity;
import com.itheima.model.behavior.pojos.UserBehavior;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 排行榜计算定时任务
 * 定期计算图书热度和用户排名，更新到 Redis
 */
@Component
@Slf4j
public class RankingCalculationTask {

    @Autowired
    private MongoTemplate mongoTemplate;

    @Autowired
    private BookPopularityRepository bookPopularityRepository;

    @Autowired
    private CacheService cacheService;

    // 热度计算权重
    private static final double WEIGHT_VIEW = 0.1;           // 浏览权重
    private static final double WEIGHT_READING = 1.0;        // 阅读权重
    private static final double WEIGHT_LIKE = 2.0;           // 点赞权重
    private static final double WEIGHT_COLLECT = 3.0;        // 收藏权重
    private static final double WEIGHT_NOTE = 2.5;           // 笔记权重
    private static final double WEIGHT_REVIEW = 3.5;         // 书评权重
    
    // 时间衰减因子（用于降低旧数据的权重）
    private static final double TIME_DECAY_FACTOR = 0.95;    // 每天衰减 5%
    private static final int TIME_DECAY_DAYS = 30;           // 衰减窗口 30天

    /**
     * 每15分钟执行一次：计算实时排行榜
     * cron: 每小时的第0、15、30、45分钟执行
     */
    @Scheduled(cron = "0 */15 * * * ?")
    public void calculateHourlyRanking() {
        try {
            log.info("========== 开始执行小时排行榜计算 ==========");
            long startTime = System.currentTimeMillis();

            // 计算图书热度排行榜
            calculateBookHotRanking();

            // 计算图书阅读排行榜
            calculateBookReadRanking();

            // 计算图书收藏排行榜
            calculateBookCollectRanking();

            // 计算图书点赞排行榜
            calculateBookLikeRanking();

            long cost = System.currentTimeMillis() - startTime;
            log.info("========== 小时排行榜计算完成，耗时: {}ms ==========", cost);

        } catch (Exception e) {
            log.error("小时排行榜计算失败", e);
        }
    }

    /**
     * 每天凌晨2点执行：计算每日排行榜
     * cron: 每天凌晨2点执行（避开凌晨0点高峰）
     */
    @Scheduled(cron = "0 0 2 * * ?")
    public void calculateDailyRanking() {
        try {
            log.info("========== 开始执行每日排行榜计算 ==========");
            long startTime = System.currentTimeMillis();

            // 计算图书热度排行榜（带时间衰减）
            calculateBookHotRankingWithDecay();

            // 计算用户阅读时长排行榜
            calculateUserReadingTimeRanking();

            // 计算用户活跃度排行榜
            calculateUserActiveRanking();

            // 持久化排行榜快照到 MongoDB
            saveRankingSnapshot();

            long cost = System.currentTimeMillis() - startTime;
            log.info("========== 每日排行榜计算完成，耗时: {}ms ==========", cost);

        } catch (Exception e) {
            log.error("每日排行榜计算失败", e);
        }
    }

    /**
     * 计算图书热度排行榜
     * 热度公式: 浏览*0.1 + 阅读*1.0 + 点赞*2.0 + 收藏*3.0 + 笔记*2.5 + 书评*3.5
     */
    private void calculateBookHotRanking() {
        try {
            log.info("开始计算图书热度排行榜...");

            // 获取所有图书ID
            Query query = new Query(Criteria.where("targetType").is("book"));
            List<String> bookIds = mongoTemplate.findDistinct(query, "targetId", UserBehavior.class, String.class);

            log.info("需要计算热度的图书数: {}", bookIds.size());

            int successCount = 0;
            Map<Integer, Double> bookScores = new HashMap<>();

            for (String bookIdStr : bookIds) {
                try {
                    // 跳过 null、空字符串或非纯数字的 ID
                    if (bookIdStr == null || bookIdStr.trim().isEmpty() || "null".equals(bookIdStr)
                            || !bookIdStr.matches("^\\d+$")) {
                        log.debug("跳过无效的 bookId: {}", bookIdStr);
                        continue;
                    }
                    
                    Integer bookId = Integer.parseInt(bookIdStr);
                    
                    // 统计各项指标
                    BookMetrics metrics = calculateBookMetrics(bookIdStr);
                    
                    // 计算热度分数
                    double hotScore = calculateHotScore(metrics);
                    
                    bookScores.put(bookId, hotScore);
                    successCount++;

                } catch (NumberFormatException e) {
                    log.warn("图书ID格式错误: {}", bookIdStr);
                } catch (Exception e) {
                    log.warn("计算图书热度失败: bookId={}", bookIdStr, e);
                }
            }

            // 批量更新到 Redis ZSet
            if (!bookScores.isEmpty()) {
                String rankingKey = KafkaConfig.Keys.BOOK_HOT_RANKING;
                // 清空旧数据
                cacheService.delete(rankingKey);
                // 批量添加新数据
                for (Map.Entry<Integer, Double> entry : bookScores.entrySet()) {
                    cacheService.zAdd(rankingKey, String.valueOf(entry.getKey()), entry.getValue());
                }
                // 只保留前1000名
                cacheService.zRemoveRange(rankingKey, 0, -1001);
                
                log.info("图书热度排行榜更新成功: 成功={}, 排行榜大小={}", 
                        successCount, cacheService.zSize(rankingKey));
            }

        } catch (Exception e) {
            log.error("计算图书热度排行榜失败", e);
        }
    }

    /**
     * 计算图书热度排行榜（带时间衰减）
     * 考虑时间因素，越新的数据权重越高
     */
    private void calculateBookHotRankingWithDecay() {
        try {
            log.info("开始计算图书热度排行榜（带时间衰减）...");

            Query query = new Query(Criteria.where("targetType").is("book"));
            List<String> bookIds = mongoTemplate.findDistinct(query, "targetId", UserBehavior.class, String.class);

            Map<Integer, Double> bookScores = new HashMap<>();

            for (String bookIdStr : bookIds) {
                try {
                    // 跳过 null、空字符串或非纯数字的 ID
                    if (bookIdStr == null || bookIdStr.trim().isEmpty() || "null".equals(bookIdStr)
                            || !bookIdStr.matches("^\\d+$")) {
                        log.debug("跳过无效的 bookId: {}", bookIdStr);
                        continue;
                    }
                    
                    Integer bookId = Integer.parseInt(bookIdStr);
                    
                    // 按天分组统计
                    Map<Integer, BookMetrics> dailyMetrics = calculateDailyBookMetrics(bookIdStr, TIME_DECAY_DAYS);
                    
                    // 计算带时间衰减的总分
                    double totalScore = 0.0;
                    for (Map.Entry<Integer, BookMetrics> entry : dailyMetrics.entrySet()) {
                        int daysAgo = entry.getKey();
                        BookMetrics metrics = entry.getValue();
                        
                        // 计算基础分数
                        double baseScore = calculateHotScore(metrics);
                        
                        // 应用时间衰减
                        double decayFactor = Math.pow(TIME_DECAY_FACTOR, daysAgo);
                        double decayedScore = baseScore * decayFactor;
                        
                        totalScore += decayedScore;
                    }
                    
                    bookScores.put(bookId, totalScore);

                } catch (NumberFormatException e) {
                    log.warn("无效的图书ID格式: bookId={}", bookIdStr);
                } catch (Exception e) {
                    log.warn("计算图书热度失败: bookId={}", bookIdStr, e);
                }
            }

            // 更新到 Redis
            if (!bookScores.isEmpty()) {
                String rankingKey = KafkaConfig.Keys.BOOK_HOT_RANKING;
                cacheService.delete(rankingKey);
                for (Map.Entry<Integer, Double> entry : bookScores.entrySet()) {
                    cacheService.zAdd(rankingKey, String.valueOf(entry.getKey()), entry.getValue());
                }
                cacheService.zRemoveRange(rankingKey, 0, -1001);
                
                log.info("图书热度排行榜（带时间衰减）更新成功: size={}", bookScores.size());
            }

        } catch (Exception e) {
            log.error("计算图书热度排行榜（带时间衰减）失败", e);
        }
    }

    /**
     * 计算图书阅读排行榜
     */
    private void calculateBookReadRanking() {
        try {
            log.info("开始计算图书阅读排行榜...");

            Query query = new Query(Criteria.where("targetType").is("book"));
            List<String> bookIds = mongoTemplate.findDistinct(query, "targetId", UserBehavior.class, String.class);

            Map<Integer, Integer> bookReadCounts = new HashMap<>();

            for (String bookIdStr : bookIds) {
                try {
                    // 跳过 null、空字符串或非纯数字的 ID
                    if (bookIdStr == null || bookIdStr.trim().isEmpty() || "null".equals(bookIdStr)
                            || !bookIdStr.matches("^\\d+$")) {
                        log.debug("跳过无效的 bookId: {}", bookIdStr);
                        continue;
                    }
                    
                    Integer bookId = Integer.parseInt(bookIdStr);
                    
                    // 统计阅读人数（去重）
                    Query readQuery = new Query(Criteria.where("targetType").is("book")
                            .and("targetId").is(bookIdStr)
                            .and("behaviorType").in(Arrays.asList("reading", "reading_start", "reading_end")));
                    
                    List<Integer> userIds = mongoTemplate.findDistinct(readQuery, "userId", UserBehavior.class, Integer.class);
                    int readCount = userIds.size();
                    
                    bookReadCounts.put(bookId, readCount);

                } catch (NumberFormatException e) {
                    log.warn("无效的图书ID格式: bookId={}", bookIdStr);
                } catch (Exception e) {
                    log.warn("计算图书阅读数失败: bookId={}", bookIdStr, e);
                }
            }

            // 更新到 Redis
            if (!bookReadCounts.isEmpty()) {
                String rankingKey = KafkaConfig.Keys.BOOK_READ_RANKING;
                cacheService.delete(rankingKey);
                for (Map.Entry<Integer, Integer> entry : bookReadCounts.entrySet()) {
                    cacheService.zAdd(rankingKey, String.valueOf(entry.getKey()), entry.getValue().doubleValue());
                }
                cacheService.zRemoveRange(rankingKey, 0, -1001);
                
                log.info("图书阅读排行榜更新成功: size={}", bookReadCounts.size());
            }

        } catch (Exception e) {
            log.error("计算图书阅读排行榜失败", e);
        }
    }

    /**
     * 计算图书收藏排行榜
     */
    private void calculateBookCollectRanking() {
        try {
            log.info("开始计算图书收藏排行榜...");

            Query query = new Query(Criteria.where("targetType").is("book"));
            List<String> bookIds = mongoTemplate.findDistinct(query, "targetId", UserBehavior.class, String.class);

            Map<Integer, Long> bookCollectCounts = new HashMap<>();

            for (String bookIdStr : bookIds) {
                try {
                    // 跳过 null、空字符串或非纯数字的 ID
                    if (bookIdStr == null || bookIdStr.trim().isEmpty() || "null".equals(bookIdStr)
                            || !bookIdStr.matches("^\\d+$")) {
                        log.debug("跳过无效的 bookId: {}", bookIdStr);
                        continue;
                    }
                    
                    Integer bookId = Integer.parseInt(bookIdStr);
                    
                    // 统计收藏数
                    long collectCount = mongoTemplate.count(
                            new Query(Criteria.where("targetType").is("book")
                                    .and("targetId").is(bookIdStr)
                                    .and("behaviorType").is("collect")),
                            UserBehavior.class);
                    
                    bookCollectCounts.put(bookId, collectCount);

                } catch (NumberFormatException e) {
                    log.warn("无效的图书ID格式: bookId={}", bookIdStr);
                } catch (Exception e) {
                    log.warn("计算图书收藏数失败: bookId={}", bookIdStr, e);
                }
            }

            // 更新到 Redis
            if (!bookCollectCounts.isEmpty()) {
                String rankingKey = KafkaConfig.Keys.BOOK_COLLECT_RANKING;
                cacheService.delete(rankingKey);
                for (Map.Entry<Integer, Long> entry : bookCollectCounts.entrySet()) {
                    cacheService.zAdd(rankingKey, String.valueOf(entry.getKey()), entry.getValue().doubleValue());
                }
                cacheService.zRemoveRange(rankingKey, 0, -1001);
                
                log.info("图书收藏排行榜更新成功: size={}", bookCollectCounts.size());
            }

        } catch (Exception e) {
            log.error("计算图书收藏排行榜失败", e);
        }
    }

    /**
     * 计算图书点赞排行榜
     */
    private void calculateBookLikeRanking() {
        try {
            log.info("开始计算图书点赞排行榜...");

            Query query = new Query(Criteria.where("targetType").is("book"));
            List<String> bookIds = mongoTemplate.findDistinct(query, "targetId", UserBehavior.class, String.class);

            Map<Integer, Long> bookLikeCounts = new HashMap<>();

            for (String bookIdStr : bookIds) {
                try {
                    // 跳过 null、空字符串或非纯数字的 ID
                    if (bookIdStr == null || bookIdStr.trim().isEmpty() || "null".equals(bookIdStr)
                            || !bookIdStr.matches("^\\d+$")) {
                        log.debug("跳过无效的 bookId: {}", bookIdStr);
                        continue;
                    }
                    
                    Integer bookId = Integer.parseInt(bookIdStr);
                    
                    // 统计点赞数
                    long likeCount = mongoTemplate.count(
                            new Query(Criteria.where("targetType").is("book")
                                    .and("targetId").is(bookIdStr)
                                    .and("behaviorType").is("like")),
                            UserBehavior.class);
                    
                    bookLikeCounts.put(bookId, likeCount);

                } catch (NumberFormatException e) {
                    log.warn("无效的图书ID格式: bookId={}", bookIdStr);
                } catch (Exception e) {
                    log.warn("计算图书点赞数失败: bookId={}", bookIdStr, e);
                }
            }

            // 更新到 Redis
            if (!bookLikeCounts.isEmpty()) {
                String rankingKey = KafkaConfig.Keys.BOOK_LIKE_RANKING;
                cacheService.delete(rankingKey);
                for (Map.Entry<Integer, Long> entry : bookLikeCounts.entrySet()) {
                    cacheService.zAdd(rankingKey, String.valueOf(entry.getKey()), entry.getValue().doubleValue());
                }
                cacheService.zRemoveRange(rankingKey, 0, -1001);
                
                log.info("图书点赞排行榜更新成功: size={}", bookLikeCounts.size());
            }

        } catch (Exception e) {
            log.error("计算图书点赞排行榜失败", e);
        }
    }

    /**
     * 计算用户阅读时长排行榜
     */
    private void calculateUserReadingTimeRanking() {
        try {
            log.info("开始计算用户阅读时长排行榜...");

            // 获取所有用户ID
            List<Integer> userIds = mongoTemplate.findDistinct(
                    new Query(), "userId", UserBehavior.class, Integer.class);

            Map<Integer, Integer> userReadingTimes = new HashMap<>();

            for (Integer userId : userIds) {
                try {
                    // 统计阅读时长（秒）
                    Query query = new Query(Criteria.where("userId").is(userId)
                            .and("behaviorType").is("reading_end"));
                    
                    List<UserBehavior> behaviors = mongoTemplate.find(query, UserBehavior.class);
                    int totalTime = behaviors.stream()
                            .mapToInt(b -> b.getDuration() != null ? b.getDuration() : 0)
                            .sum();
                    
                    // 转换为分钟
                    int timeInMinutes = totalTime / 60;
                    
                    if (timeInMinutes > 0) {
                        userReadingTimes.put(userId, timeInMinutes);
                    }

                } catch (Exception e) {
                    log.warn("计算用户阅读时长失败: userId={}", userId, e);
                }
            }

            // 更新到 Redis
            if (!userReadingTimes.isEmpty()) {
                String rankingKey = KafkaConfig.Keys.USER_READING_TIME_RANKING;
                cacheService.delete(rankingKey);
                for (Map.Entry<Integer, Integer> entry : userReadingTimes.entrySet()) {
                    cacheService.zAdd(rankingKey, String.valueOf(entry.getKey()), entry.getValue().doubleValue());
                }
                cacheService.zRemoveRange(rankingKey, 0, -1001);
                
                log.info("用户阅读时长排行榜更新成功: size={}", userReadingTimes.size());
            }

        } catch (Exception e) {
            log.error("计算用户阅读时长排行榜失败", e);
        }
    }

    /**
     * 计算用户活跃度排行榜
     */
    private void calculateUserActiveRanking() {
        try {
            log.info("开始计算用户活跃度排行榜...");

            List<Integer> userIds = mongoTemplate.findDistinct(
                    new Query(), "userId", UserBehavior.class, Integer.class);

            Map<Integer, Long> userActiveCounts = new HashMap<>();

            for (Integer userId : userIds) {
                try {
                    // 统计行为次数
                    long count = mongoTemplate.count(
                            new Query(Criteria.where("userId").is(userId)),
                            UserBehavior.class);
                    
                    userActiveCounts.put(userId, count);

                } catch (Exception e) {
                    log.warn("计算用户活跃度失败: userId={}", userId, e);
                }
            }

            // 更新到 Redis
            if (!userActiveCounts.isEmpty()) {
                String rankingKey = KafkaConfig.Keys.USER_ACTIVE_RANKING;
                cacheService.delete(rankingKey);
                for (Map.Entry<Integer, Long> entry : userActiveCounts.entrySet()) {
                    cacheService.zAdd(rankingKey, String.valueOf(entry.getKey()), entry.getValue().doubleValue());
                }
                cacheService.zRemoveRange(rankingKey, 0, -1001);
                
                log.info("用户活跃度排行榜更新成功: size={}", userActiveCounts.size());
            }

        } catch (Exception e) {
            log.error("计算用户活跃度排行榜失败", e);
        }
    }

    /**
     * 计算图书的各项指标
     */
    private BookMetrics calculateBookMetrics(String bookId) {
        BookMetrics metrics = new BookMetrics();

        // 浏览次数
        metrics.viewCount = (int) mongoTemplate.count(
                new Query(Criteria.where("targetType").is("book")
                        .and("targetId").is(bookId)
                        .and("behaviorType").in(Arrays.asList("book_view", "page_view"))),
                UserBehavior.class);

        // 阅读人数（去重）
        metrics.readingCount = mongoTemplate.findDistinct(
                new Query(Criteria.where("targetType").is("book")
                        .and("targetId").is(bookId)
                        .and("behaviorType").in(Arrays.asList("reading", "reading_start"))),
                "userId", UserBehavior.class, Integer.class).size();

        // 点赞数
        metrics.likeCount = (int) mongoTemplate.count(
                new Query(Criteria.where("targetType").is("book")
                        .and("targetId").is(bookId)
                        .and("behaviorType").is("like")),
                UserBehavior.class);

        // 收藏数
        metrics.collectCount = (int) mongoTemplate.count(
                new Query(Criteria.where("targetType").is("book")
                        .and("targetId").is(bookId)
                        .and("behaviorType").is("collect")),
                UserBehavior.class);

        // 笔记数
        metrics.noteCount = (int) mongoTemplate.count(
                new Query(Criteria.where("targetType").is("book")
                        .and("targetId").is(bookId)
                        .and("behaviorType").is("note_create")),
                UserBehavior.class);

        // 书评数
        metrics.reviewCount = (int) mongoTemplate.count(
                new Query(Criteria.where("targetType").is("book")
                        .and("targetId").is(bookId)
                        .and("behaviorType").is("review_create")),
                UserBehavior.class);

        return metrics;
    }

    /**
     * 按天计算图书指标
     */
    private Map<Integer, BookMetrics> calculateDailyBookMetrics(String bookId, int days) {
        Map<Integer, BookMetrics> dailyMetrics = new HashMap<>();
        
        LocalDateTime now = LocalDateTime.now();
        
        for (int i = 0; i < days; i++) {
            LocalDateTime startOfDay = now.minusDays(i).toLocalDate().atStartOfDay();
            LocalDateTime endOfDay = startOfDay.plusDays(1);
            
            Date startDate = Date.from(startOfDay.atZone(ZoneId.systemDefault()).toInstant());
            Date endDate = Date.from(endOfDay.atZone(ZoneId.systemDefault()).toInstant());
            
            BookMetrics metrics = new BookMetrics();
            
            // 当天的各项指标 - 每个查询都创建新的 Criteria 对象
            metrics.viewCount = (int) mongoTemplate.count(
                    new Query(Criteria.where("targetType").is("book")
                            .and("targetId").is(bookId)
                            .and("createTime").gte(startDate).lt(endDate)
                            .and("behaviorType").in(Arrays.asList("book_view", "page_view"))),
                    UserBehavior.class);
            
            metrics.readingCount = mongoTemplate.findDistinct(
                    new Query(Criteria.where("targetType").is("book")
                            .and("targetId").is(bookId)
                            .and("createTime").gte(startDate).lt(endDate)
                            .and("behaviorType").in(Arrays.asList("reading", "reading_start"))),
                    "userId", UserBehavior.class, Integer.class).size();
            
            metrics.likeCount = (int) mongoTemplate.count(
                    new Query(Criteria.where("targetType").is("book")
                            .and("targetId").is(bookId)
                            .and("createTime").gte(startDate).lt(endDate)
                            .and("behaviorType").is("like")),
                    UserBehavior.class);
            
            metrics.collectCount = (int) mongoTemplate.count(
                    new Query(Criteria.where("targetType").is("book")
                            .and("targetId").is(bookId)
                            .and("createTime").gte(startDate).lt(endDate)
                            .and("behaviorType").is("collect")),
                    UserBehavior.class);
            
            metrics.noteCount = (int) mongoTemplate.count(
                    new Query(Criteria.where("targetType").is("book")
                            .and("targetId").is(bookId)
                            .and("createTime").gte(startDate).lt(endDate)
                            .and("behaviorType").is("note_create")),
                    UserBehavior.class);
            
            metrics.reviewCount = (int) mongoTemplate.count(
                    new Query(Criteria.where("targetType").is("book")
                            .and("targetId").is(bookId)
                            .and("createTime").gte(startDate).lt(endDate)
                            .and("behaviorType").is("review_create")),
                    UserBehavior.class);
            
            dailyMetrics.put(i, metrics);
        }
        
        return dailyMetrics;
    }

    /**
     * 计算热度分数
     * 公式: 浏览*0.1 + 阅读*1.0 + 点赞*2.0 + 收藏*3.0 + 笔记*2.5 + 书评*3.5
     */
    private double calculateHotScore(BookMetrics metrics) {
        return metrics.viewCount * WEIGHT_VIEW
                + metrics.readingCount * WEIGHT_READING
                + metrics.likeCount * WEIGHT_LIKE
                + metrics.collectCount * WEIGHT_COLLECT
                + metrics.noteCount * WEIGHT_NOTE
                + metrics.reviewCount * WEIGHT_REVIEW;
    }

    /**
     * 持久化排行榜快照到 MongoDB
     */
    private void saveRankingSnapshot() {
        try {
            log.info("开始保存排行榜快照...");
            
            // TODO: 可以将当前排行榜数据保存到 MongoDB，用于历史趋势分析
            // 例如：保存 Top100 的数据，记录排名变化
            
            log.info("排行榜快照保存完成");
            
        } catch (Exception e) {
            log.error("保存排行榜快照失败", e);
        }
    }

    /**
     * 图书指标内部类
     */
    private static class BookMetrics {
        int viewCount = 0;
        int readingCount = 0;
        int likeCount = 0;
        int collectCount = 0;
        int noteCount = 0;
        int reviewCount = 0;
    }
}


